/**
 * @p.md
 * 
 * PostgreSQLProvider 是一个基于 PostgreSQL 数据库的持久化提供者
 * 
 * 注意：使用此提供者需要安装 pg 依赖:
 * ```
 * npm install pg
 * ```
 */
import { Pool, PoolClient, QueryConfig } from 'pg';
import { PersistenceProvider } from './persistenceProvider';

/**
 * PostgreSQL 持久化提供者，用于将事件和快照持久化到 PostgreSQL 数据库
 */
export class PostgreSQLProvider implements PersistenceProvider {
  private pool: Pool;
  private initialized: boolean = false;
  
  /**
   * 构造函数
   * @param connectionString PostgreSQL 连接字符串，默认为 'postgresql://postgres:postgres@localhost:5432/protoactor'
   */
  constructor(private readonly connectionString: string = 'postgresql://postgres:postgres@localhost:5432/protoactor') {
    this.pool = new Pool({ connectionString });
    this.initialize();
  }
  
  /**
   * 初始化数据库表
   */
  private async initialize(): Promise<void> {
    if (this.initialized) return;
    
    const client = await this.pool.connect();
    try {
      // 开启事务
      await client.query('BEGIN');
      
      // 创建事件表
      await client.query(`
        CREATE TABLE IF NOT EXISTS events (
          persistence_id TEXT NOT NULL,
          event_index INTEGER NOT NULL,
          event_data JSONB NOT NULL,
          timestamp BIGINT NOT NULL,
          PRIMARY KEY (persistence_id, event_index)
        )
      `);
      
      // 创建快照表
      await client.query(`
        CREATE TABLE IF NOT EXISTS snapshots (
          persistence_id TEXT NOT NULL,
          snapshot_index INTEGER NOT NULL,
          snapshot_data JSONB NOT NULL,
          timestamp BIGINT NOT NULL,
          PRIMARY KEY (persistence_id, snapshot_index)
        )
      `);
      
      // 创建索引
      await client.query(`
        CREATE INDEX IF NOT EXISTS idx_events_persistence_id ON events (persistence_id)
      `);
      
      await client.query(`
        CREATE INDEX IF NOT EXISTS idx_snapshots_persistence_id ON snapshots (persistence_id)
      `);
      
      // 提交事务
      await client.query('COMMIT');
      
      console.log(`已连接到 PostgreSQL: ${this.connectionString}`);
      this.initialized = true;
    } catch (error) {
      // 回滚事务
      await client.query('ROLLBACK');
      console.error('PostgreSQL 初始化错误:', error);
      throw error;
    } finally {
      // 释放客户端回连接池
      client.release();
    }
  }
  
  /**
   * 关闭数据库连接池
   */
  public async close(): Promise<void> {
    await this.pool.end();
    console.log('PostgreSQL 连接已关闭');
  }
  
  /**
   * 执行查询，确保初始化
   * @param callback 回调函数，接收客户端作为参数
   */
  private async withClient<T>(callback: (client: PoolClient) => Promise<T>): Promise<T> {
    await this.ensureInitialized();
    
    const client = await this.pool.connect();
    try {
      return await callback(client);
    } finally {
      client.release();
    }
  }
  
  /**
   * 确保数据库已初始化
   */
  private async ensureInitialized(): Promise<void> {
    if (!this.initialized) {
      await this.initialize();
    }
  }
  
  /**
   * 获取指定持久化 ID 的事件
   * @param persistenceId 持久化 ID
   * @param fromIndex 开始索引（包含）
   * @param toIndex 结束索引（包含）
   * @returns 事件数组
   */
  async getEventsByPersistenceId(persistenceId: string, fromIndex: number, toIndex: number): Promise<any[]> {
    return this.withClient(async (client) => {
      const query: QueryConfig = {
        text: `SELECT event_data FROM events 
               WHERE persistence_id = $1 AND event_index >= $2 AND event_index <= $3
               ORDER BY event_index ASC`,
        values: [persistenceId, fromIndex, toIndex]
      };
      
      const result = await client.query(query);
      return result.rows.map(row => row.event_data);
    });
  }
  
  /**
   * 获取指定持久化 ID 的最新快照
   * @param persistenceId 持久化 ID
   * @returns 快照数据，如果不存在则返回 null
   */
  async getSnapshotByPersistenceId(persistenceId: string): Promise<any | null> {
    return this.withClient(async (client) => {
      const query: QueryConfig = {
        text: `SELECT snapshot_data FROM snapshots 
               WHERE persistence_id = $1 
               ORDER BY snapshot_index DESC 
               LIMIT 1`,
        values: [persistenceId]
      };
      
      const result = await client.query(query);
      return result.rows.length > 0 ? result.rows[0].snapshot_data : null;
    });
  }
  
  /**
   * 持久化事件
   * @param persistenceId 持久化 ID
   * @param eventIndex 事件索引
   * @param event 事件数据
   */
  async persistEvent(persistenceId: string, eventIndex: number, event: any): Promise<void> {
    return this.withClient(async (client) => {
      const timestamp = Date.now();
      
      const query: QueryConfig = {
        text: `INSERT INTO events (persistence_id, event_index, event_data, timestamp)
               VALUES ($1, $2, $3, $4)
               ON CONFLICT (persistence_id, event_index) 
               DO UPDATE SET event_data = $3, timestamp = $4`,
        values: [persistenceId, eventIndex, event, timestamp]
      };
      
      await client.query(query);
    });
  }
  
  /**
   * 持久化快照
   * @param persistenceId 持久化 ID
   * @param snapshotIndex 快照索引
   * @param snapshot 快照数据
   */
  async persistSnapshot(persistenceId: string, snapshotIndex: number, snapshot: any): Promise<void> {
    return this.withClient(async (client) => {
      const timestamp = Date.now();
      
      const query: QueryConfig = {
        text: `INSERT INTO snapshots (persistence_id, snapshot_index, snapshot_data, timestamp)
               VALUES ($1, $2, $3, $4)
               ON CONFLICT (persistence_id, snapshot_index) 
               DO UPDATE SET snapshot_data = $3, timestamp = $4`,
        values: [persistenceId, snapshotIndex, snapshot, timestamp]
      };
      
      await client.query(query);
    });
  }
  
  /**
   * 删除事件
   * @param persistenceId 持久化 ID
   * @param fromIndex 开始索引（包含）
   * @param toIndex 结束索引（包含）
   */
  async deleteEvents(persistenceId: string, fromIndex: number, toIndex: number): Promise<void> {
    return this.withClient(async (client) => {
      const query: QueryConfig = {
        text: `DELETE FROM events 
               WHERE persistence_id = $1 AND event_index >= $2 AND event_index <= $3`,
        values: [persistenceId, fromIndex, toIndex]
      };
      
      await client.query(query);
    });
  }
  
  /**
   * 删除快照
   * @param persistenceId 持久化 ID
   * @param fromIndex 开始索引（包含）
   * @param toIndex 结束索引（包含）
   */
  async deleteSnapshots(persistenceId: string, fromIndex: number, toIndex: number): Promise<void> {
    return this.withClient(async (client) => {
      const query: QueryConfig = {
        text: `DELETE FROM snapshots 
               WHERE persistence_id = $1 AND snapshot_index >= $2 AND snapshot_index <= $3`,
        values: [persistenceId, fromIndex, toIndex]
      };
      
      await client.query(query);
    });
  }
} 